import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
/**
 * 自定义分区
 */
public class PartitionProducer
{
    public static void main(String[] args) {

        String server = "192.168.60.88:9092";
        // 1.创建配置信息
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, server);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 添加分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "MyPartitioner");

        // 2.创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // 3.发送数据
        producer.send(new ProducerRecord<>("test2", "fmy","这是带回调方法的生产者发送的消息!"), (RecordMetadata metadata, Exception exception)-> {
            if (exception == null) {
                System.out.println("已收到ack,这里是回调方法");
                System.out.println("元数据分区:"+metadata.partition() + ",偏移量:" + metadata.offset());
            } else {
                exception.printStackTrace();
            }
        });

        // 4.关闭资源
        producer.close();
    }
}
